package com.monitor.zk.zookeeper.core.mq;

import bean.RocketMqConfig;
import com.monitor.zk.zookeeper.handle.PrepareHandleProcess;
import com.monitor.zk.zookeeper.handle.RUN_STATUS;
import com.monitor.zk.zookeeper.handle.SendThreadHandleProcess;
import core.ParamsContext;
import core.annotation.StartHandle;
import core.annotation.AfterStart;
import core.scan.ScanXmlConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

@AfterStart
@Slf4j
@Component
public class Producer implements StartHandle {

    private static DefaultMQProducer producer = null;
    private static String topic = null;
    private static String relativeIp;

    /**
     * 初始化
     * @throws Exception
     */
    public void initDefaultMQProducer(ParamsContext paramsContext) throws Exception {
        relativeIp = new String(CuratorFrameworkFactory.getLocalAddress(),RemotingHelper.DEFAULT_CHARSET);
        producer = new DefaultMQProducer();
        RocketMqConfig config = (RocketMqConfig) ScanXmlConfig.newInScanXmlConfig().loadConfig("rocketmq", RocketMqConfig.class);
        producer.setInstanceName(config.getGroupName());
        producer.setProducerGroup(config.getGroupName());
        producer.setVipChannelEnabled(false);
        producer.setClientIP(config.getClientIP());
        producer.setNamesrvAddr(config.getNamesrvAddr());
        producer.setRetryTimesWhenSendFailed(config.getRetryTimes());
        producer.setInstanceName(config.getGroupName());
        producer.setSendMsgTimeout(config.getTimeOut());
        producer.start();
        topic = config.getTopic();
        log.info("rocketMq start successful");
        changeHandleStartStatus();
    }

    public void changeHandleStartStatus(){
        if(PrepareHandleProcess.STATUS == RUN_STATUS.STOP){
            PrepareHandleProcess.STATUS = RUN_STATUS.RUN;
        }
        if(SendThreadHandleProcess.STATUS == RUN_STATUS.STOP){
            SendThreadHandleProcess.STATUS = RUN_STATUS.RUN;
        }
    }

    public static void send(String msg,String tag) throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
        asynSend(msg,tag,null);
    }

    public static void asynSend(String msg,String tag,SendCallback sendCallback) throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message();
        message.setTags(tag);
        message.setBody(msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
        message.putUserProperty("relativeIp",relativeIp);
        message.setTopic(topic);
        log.info("发送数据:{}",message.toString());
        if(sendCallback == null){
            producer.send(message);
        }else{
            producer.send(message,sendCallback);
        }

    }

    public static void sendDefaultCallBack(String msg,String tag) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        asynSend(msg,tag,new SendCallBackListen());
    }


    @Override
    public void start(ParamsContext paramsContext) throws Exception {
        initDefaultMQProducer(paramsContext);
    }
}
